using EasyNetQ.Internals;

namespace EasyNetQ;

/// <summary>
///     Various generic extensions for <see cref="IPubSub"/>
/// </summary>
public static class PubSubExtensions
{
    /// <summary>
    /// Publishes a message with a topic.
    /// When used with publisher confirms the task completes when the publish is confirmed.
    /// Task will throw an exception if the confirm is NACK'd or times out.
    /// </summary>
    /// <typeparam name="T">The message type</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="message">The message to publish</param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns></returns>
    public static Task PublishAsync<T>(this IPubSub pubSub, T message, CancellationToken cancellationToken = default)
        => pubSub.PublishAsync(message, _ => { }, cancellationToken);

    /// <summary>
    /// Publishes a message with a topic.
    /// When used with publisher confirms the task completes when the publish is confirmed.
    /// Task will throw an exception if the confirm is NACK'd or times out.
    /// </summary>
    /// <typeparam name="T">The message type</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// /// <param name="message">The message to publish</param>
    /// <param name="topic">The topic string</param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns></returns>
    public static Task PublishAsync<T>(this IPubSub pubSub, T message, string topic, CancellationToken cancellationToken = default)
        => pubSub.PublishAsync(message, c => c.WithTopic(topic), cancellationToken);

    /// <summary>
    /// Publishes a message.
    /// </summary>
    /// <typeparam name="T">The message type</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="message">The message to publish</param>
    /// <param name="cancellationToken">The cancellation token</param>
    public static void Publish<T>(this IPubSub pubSub, T message, CancellationToken cancellationToken = default)
        => pubSub.Publish(message, _ => { }, cancellationToken);

    /// <summary>
    /// Publishes a message.
    /// </summary>
    /// <typeparam name="T">The message type</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="message">The message to publish</param>
    /// <param name="configure">
    /// Fluent configuration e.g. x => x.WithTopic("*.brighton").WithPriority(2)
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    public static void Publish<T>(this IPubSub pubSub, T message, Action<IPublishConfiguration> configure, CancellationToken cancellationToken = default)
    {
        pubSub.PublishAsync(message, configure, cancellationToken)
            .GetAwaiter()
            .GetResult();
    }

    /// <summary>
    /// Publishes a message with a topic
    /// </summary>
    /// <typeparam name="T">The message type</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="message">The message to publish</param>
    /// <param name="topic">The topic string</param>
    /// <param name="cancellationToken">The cancellation token</param>
    public static void Publish<T>(this IPubSub pubSub, T message, string topic, CancellationToken cancellationToken = default)
        => pubSub.Publish(message, c => c.WithTopic(topic), cancellationToken);

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. When onMessage completes the message
    /// receipt is Ack'd. All onMessage delegates are processed on a single thread so you should
    /// avoid long running blocking IO operations. Consider using SubscribeAsync
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static Task<SubscriptionResult> SubscribeAsync<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Action<T> onMessage,
        CancellationToken cancellationToken = default
    )
    {
        return pubSub.SubscribeAsync(
            subscriptionId,
            onMessage,
            _ => { },
            cancellationToken
        );
    }

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. When onMessage completes the message
    /// receipt is Ack'd. All onMessage delegates are processed on a single thread so you should
    /// avoid long running blocking IO operations. Consider using SubscribeAsync
    /// </param>
    /// <param name="configure">
    /// Fluent configuration e.g. x => x.WithTopic("uk.london")
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static Task<SubscriptionResult> SubscribeAsync<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Action<T> onMessage,
        Action<ISubscriptionConfiguration> configure,
        CancellationToken cancellationToken = default
    )
    {


        var onMessageAsync = TaskHelpers.FromAction<T>((m, _) => onMessage(m));

        return pubSub.SubscribeAsync(
            subscriptionId,
            onMessageAsync,
            configure,
            cancellationToken
        );
    }

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// Allows the subscriber to complete asynchronously.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. onMessage can immediately return a Task and
    /// then continue processing asynchronously. When the Task completes the message will be
    /// Ack'd.
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static Task<SubscriptionResult> SubscribeAsync<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Func<T, Task> onMessage,
        CancellationToken cancellationToken = default
    )
    {
        return pubSub.SubscribeAsync<T>(
            subscriptionId,
            (m, _) => onMessage(m),
            _ => { },
            cancellationToken
        );
    }

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. When onMessage completes the message
    /// receipt is Ack'd. All onMessage delegates are processed on a single thread so you should
    /// avoid long running blocking IO operations. Consider using SubscribeAsync
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static SubscriptionResult Subscribe<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Action<T> onMessage,
        CancellationToken cancellationToken = default
    )
    {
        return pubSub.Subscribe(
            subscriptionId,
            onMessage,
            _ => { },
            cancellationToken
        );
    }

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. When onMessage completes the message
    /// receipt is Ack'd. All onMessage delegates are processed on a single thread so you should
    /// avoid long running blocking IO operations. Consider using SubscribeAsync
    /// </param>
    /// <param name="configure">
    /// Fluent configuration e.g. x => x.WithTopic("uk.london")
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static SubscriptionResult Subscribe<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Action<T> onMessage,
        Action<ISubscriptionConfiguration> configure,
        CancellationToken cancellationToken = default
    )
    {
        var onMessageAsync = TaskHelpers.FromAction<T>((m, _) => onMessage(m));

        return pubSub.Subscribe(
            subscriptionId,
            onMessageAsync,
            configure,
            cancellationToken
        );
    }

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// Allows the subscriber to complete asynchronously.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. onMessage can immediately return a Task and
    /// then continue processing asynchronously. When the Task completes the message will be
    /// Ack'd.
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static SubscriptionResult Subscribe<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Func<T, Task> onMessage,
        CancellationToken cancellationToken = default
    )
    {
        return pubSub.Subscribe<T>(
            subscriptionId,
            (m, _) => onMessage(m),
            _ => { },
            cancellationToken
        );
    }

    /// <summary>
    /// Subscribes to a stream of messages that match a .NET type.
    /// Allows the subscriber to complete asynchronously.
    /// </summary>
    /// <typeparam name="T">The type to subscribe to</typeparam>
    /// <param name="pubSub">The pubSub instance</param>
    /// <param name="subscriptionId">
    /// A unique identifier for the subscription. Two subscriptions with the same subscriptionId
    /// and type will get messages delivered in turn. This is useful if you want multiple subscribers
    /// to load balance a subscription in a round-robin fashion.
    /// </param>
    /// <param name="onMessage">
    /// The action to run when a message arrives. onMessage can immediately return a Task and
    /// then continue processing asynchronously. When the Task completes the message will be
    /// Ack'd.
    /// </param>
    /// <param name="configure">
    /// Fluent configuration e.g. x => x.WithTopic("uk.london")
    /// </param>
    /// <param name="cancellationToken">The cancellation token</param>
    /// <returns>
    /// An <see cref="SubscriptionResult"/>
    /// Call Dispose on it to cancel the subscription.
    /// </returns>
    public static SubscriptionResult Subscribe<T>(
        this IPubSub pubSub,
        string subscriptionId,
        Func<T, CancellationToken, Task> onMessage,
        Action<ISubscriptionConfiguration> configure,
        CancellationToken cancellationToken = default
    )
    {
        return pubSub.SubscribeAsync(subscriptionId, onMessage, configure, cancellationToken)
            .GetAwaiter()
            .GetResult();
    }
}
